package com.lizhiyu.flink.demo6_watermark;

import com.lizhiyu.flink.demo5_window.CustomSource3;
import com.lizhiyu.flink.demo5_window.CustomSource4;
import com.lizhiyu.flink.model.TimeUtils;
import com.lizhiyu.flink.model.VideoOrder;
import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkGeneratorSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;
import java.util.stream.Collectors;

/**
 *  flink分为3类时间
 *  1、数据创建的时间        名字叫做  EventTime  (重点注意)
 *  2、数据进入到flink的时间 名字叫做  ingestionTime
 *  3、flink处理数据的时间   名字叫做  ProcessingTime
 *
 * window(TumblingProcessingTimeWindows.of(Time.seconds(5)))  使用这种方法使用的是flink的处理数据的时间 ProcessingTime 进行对数据分窗口
 * 如果想要效率高，数据有不确定性可以使用 processingTime 最为分窗的时间   ()
 * 如果想要数据精准分窗要使用EventTime，防止数据乱序 (例如:统计固定时间订单数量时候使用，如果延迟比较严重，例如双11的12点屏幕，过了很长时间才显示出有卖的信息)
 *
 */
public class EventTime {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> ds = env.addSource(new CustomSource4());
        ds.setParallelism(1);
        ds.print("------------source:");
        //对数据格式进行转换
        SingleOutputStreamOperator<Tuple3<String, Integer,String>> mapDs = ds.map(new MapFunction<String, Tuple3<String, Integer,String>>() {
            @Override
            public Tuple3<String, Integer,String> map(String value) throws Exception {
                String[] split = value.split(":");
                Tuple3 Tuple3 = new Tuple3(split[0],Integer.parseInt(split[1]),split[2]);
                return Tuple3;
            }
        });

        //设置eventTime 同时设置迟到数据3秒钟
        SingleOutputStreamOperator<Tuple3<String, Integer, String>> watermarks = mapDs.assignTimestampsAndWatermarks(WatermarkStrategy
                //指定最大允许的延迟/乱序 时间
                .<Tuple3<String, Integer, String>>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                        new SerializableTimestampAssigner<Tuple3<String, Integer, String>>() {
                            @Override
                            public long extractTimestamp(Tuple3<String, Integer, String> event, long timestamp) {
                                //指定POJO的事件时间列,返回时间戳，返回这条数据对应的event time
                                long time = TimeUtils.strToDate(event.f2).getTime();
                                return time;
                            }
                        }
                ));

        //对数据进行分组聚合
        KeyedStream<Tuple3<String, Integer,String>, String> keyByDs = watermarks.keyBy(new KeySelector<Tuple3<String, Integer,String>, String>() {
            @Override
            public String getKey(Tuple3<String, Integer,String> value) throws Exception {
                return value.f0;
            }
        });


        SingleOutputStreamOperator<VideoOrder> apply = keyByDs.window(
//                .window(TumblingEventTimeWindows.of(Time.seconds(8))) //添加滚动窗口 ，采用事件时间进行处理
//                .window(SlidingProcessingTimeWindows.of(Time.seconds(10),Time.seconds(5))) //添加滑动窗口，采用处理时间进行处理
//                TumblingProcessingTimeWindows.of(Time.seconds(5))
                //设置使用数据创建时间  窗口大小 5秒
                TumblingEventTimeWindows.of(Time.seconds(5))
        )

                //<IN, OUT, KEY, W extends Window>
                .apply(new WindowFunction<Tuple3<String, Integer, String>, VideoOrder, String, TimeWindow>() {
                    @Override
                    public void apply(String key, TimeWindow window, Iterable<Tuple3<String, Integer, String>> input, Collector<VideoOrder> out) throws Exception {
                        List<Tuple3<String, Integer, String>> list = IteratorUtils.toList(input.iterator());
                        VideoOrder videoOrder = new VideoOrder();

                        videoOrder.setMoney(list.stream().collect(Collectors.summingInt(obj -> obj.f1)));
                        videoOrder.setTitle(list.get(0).f0);
                        videoOrder.setCreateTimeStr(list.get(0).f2);
                        out.collect(videoOrder);
                    }
                });
        apply.print("sink:");
        env.execute();
    }
}
